package com.heima.schedule.service.impl;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.heima.common.constants.ScheduleConstants;
import com.heima.common.redis.CacheService;
import com.heima.model.schedule.dtos.Task;
import com.heima.model.schedule.pojos.Taskinfo;
import com.heima.model.schedule.pojos.TaskinfoLogs;
import com.heima.schedule.mapper.TaskinfoLogsMapper;
import com.heima.schedule.mapper.TaskinfoMapper;
import com.heima.schedule.service.TaskService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.util.Calendar;
import java.util.Date;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ExecutorService;



@Service
@Slf4j
public class TaskServiceImpl implements TaskService {
    @Autowired
    private TaskinfoMapper taskinfoMapper;
    @Autowired
    private TaskinfoLogsMapper taskinfoLogsMapper;
    @Autowired
    private CacheService cacheService;
    @Autowired
    private ExecutorService executorService;

/**
 * 消费者将数据存入消息队列
 */

    @Override
    public long addTask(Task task){
        //将数据存入数据库
        boolean b=addTaskINDB(task);
        //如果成功就去判断是否
        if (b) {
            //存完后存入rides
            addTaskINRides(task);
        }
        return task.getTaskId();
    }

    @Override
    public Task poll(int type, int priority) {

        String key = type+"_"+priority;
        String s = cacheService.lRightPop(ScheduleConstants.TOPIC + key);

        if (s!=null){
            Task task = JSON.parseObject(s, Task.class);
            LambdaQueryWrapper<Taskinfo> taskinfoLambdaQueryWrapper = new LambdaQueryWrapper<>();
            taskinfoLambdaQueryWrapper.eq(Taskinfo::getTaskId,task.getTaskId());
            taskinfoMapper.delete(taskinfoLambdaQueryWrapper);
            TaskinfoLogs taskinfoLogs = new TaskinfoLogs();
            BeanUtils.copyProperties(task,taskinfoLogs);
            taskinfoLogs.setStatus(ScheduleConstants.CANCELLED);
            return task;
        }

        return null;
    }

    //将数据存入rides
    private void addTaskINRides(Task task) {
        //添加到rides
        //如果添加到rides
        String key = task.getTaskType()+"_"+task.getPriority();
        long executeTime = task.getExecuteTime();

        Calendar pushTime = Calendar.getInstance();
        pushTime.add(Calendar.MINUTE,5);
        long timeInMillis = pushTime.getTimeInMillis();
        if (executeTime<=System.currentTimeMillis()){
            //已经到了发布时间
            cacheService.lLeftPush(ScheduleConstants.TOPIC+key, JSON.toJSONString(task));
        }else if(executeTime <= timeInMillis){
            cacheService.zAdd(ScheduleConstants.FUTURE+key, JSON.toJSONString(task),executeTime);
        }
    }

    private boolean addTaskINDB(Task task) {
        try {
            //保存到taskinfo数据库
            Taskinfo taskinfo = new Taskinfo();
            BeanUtils.copyProperties(task,taskinfo);
            taskinfo.setExecuteTime(new Date(task.getExecuteTime()));
            taskinfoMapper.insert(taskinfo);

            task.setTaskId(taskinfo.getTaskId());

            //保存到taskinfo_log数据库
            TaskinfoLogs taskinfoLogs = new TaskinfoLogs();
            BeanUtils.copyProperties(task,taskinfoLogs);
            taskinfoLogs.setVersion(1);
            taskinfoLogs.setExecuteTime(new Date(task.getExecuteTime()));
            taskinfoLogs.setStatus(ScheduleConstants.SCHEDULED);
            taskinfoLogsMapper.insert(taskinfoLogs);
            return true;
        }catch (Exception e){
            log.error("存入错误{}",e);
            return false;
        }
    }

    /**
     * 定时器,将set中的已经到了时间的要执行的消息放入list中
     *
     */
    @Scheduled(cron = "0 0/1 * * * ? ")
    public void futureTaskSync(){
        cacheService.tryLock(ScheduleConstants.FUTURE_TASK_SYNC,1000 * 30);
        //使用管道加快rides的查询效率
        cacheService.getstringRedisTemplate().executePipelined(new RedisCallback<Object>() {
            @Override
            public Object doInRedis(RedisConnection connection) throws DataAccessException {
                //根据游标去将未来的set的key查出来
                Set<String> futureKeySet = cacheService.scan(ScheduleConstants.FUTURE+"*");

                for (String futureKey : futureKeySet) {
                    //为了优化性能将查出来的futureKey都去进行处理
/*                    executorService.submit(
                            new Runnable() {
                                @Override
                                public void run() {
                                    String[] split = futureKey.split(ScheduleConstants.FUTURE);
                                    //转到push中
                                    String topic = ScheduleConstants.TOPIC+split[1];
                                    Set<String> sortList = cacheService.zRangeByScore(futureKey, 0, System.currentTimeMillis());
                                    if(!sortList.isEmpty()){
                                        cacheService.refreshWithPipeline(futureKey,topic,sortList);
                                    }

                                }
                            }
                    );*/
                    String[] split = futureKey.split(ScheduleConstants.FUTURE);
                    //转到push中
                    String topic = ScheduleConstants.TOPIC + split[1];
                    Set<String> sortList = cacheService.zRangeByScore(futureKey, 0, System.currentTimeMillis());
                    if (!sortList.isEmpty()) {
                        cacheService.refreshWithPipeline(futureKey, topic, sortList);
                    }
                }
                return null;
            }
        });
    }

    /**
     * 先将rides中的缓存清除
     *  将数据库中小于5分钟就要去执行的消息去进行放入到set中
     *  同时也是在重启的时候进行服务重启的时候可以快速恢复,
     *
     */
    @Scheduled(cron = "0 0/5 * * * ?")
    @PostConstruct
    public void futureTaskLoad(){
        //先清空缓存
        cleanRides();
        //先获得同步到加载同步锁
        String token = cacheService.tryLock(ScheduleConstants.FUTURE_TASK_LOAD, 30);
        if (StringUtils.isBlank(token)){
            Calendar calendar = Calendar.getInstance();
            calendar.add(Calendar.MINUTE,5);
            Date after5MIN = calendar.getTime();
            LambdaQueryWrapper<Taskinfo> lambdaQueryWrapper = new LambdaQueryWrapper();
            lambdaQueryWrapper.le(Taskinfo::getExecuteTime,after5MIN);
            List<Taskinfo> taskinfoInvokeAfter5 = taskinfoMapper.selectList(lambdaQueryWrapper);
            for (Taskinfo taskinfo : taskinfoInvokeAfter5) {
                Task task = new Task();
                BeanUtils.copyProperties(taskinfo,task);
                addTaskINRides(task);

            }
        }

    }
    /**
     *清理缓存这里是为了,防止rides里的数据出现问题
     *
     */

    private void cleanRides() {
        Set<String> futurekeySet = cacheService.scan(ScheduleConstants.FUTURE);
        cacheService.delete(futurekeySet);
        Set<String> topicKeySet = cacheService.scan(ScheduleConstants.TOPIC);
        cacheService.delete(topicKeySet);
    }


}
